package com.atguigu.dga.assess.assess.security;

import ch.qos.logback.core.encoder.EchoEncoder;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.assess.assess.AssessorTemplate;
import com.atguigu.dga.assess.bean.AssessParam;
import com.atguigu.dga.assess.bean.GovernanceAssessDetail;
import com.atguigu.dga.assess.bean.GovernanceMetric;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.URI;

/*
检查该表是最高权限的目录或文件，如果文件超过{file_permission}或者目录超过{dir_permission}则给0分，其余给10分
 */
@Component("PERMISSION_CHECK")
public class CheckDirPermission extends AssessorTemplate {

    @Value("hdfs://hadoop102:8020")
    private String hdfsUri;

    @Value("${hdfs.admin}")
    private String hdfsAdmin;

    @Override
    protected void assess(AssessParam param, GovernanceAssessDetail detail) throws Exception {
        //先取参数
        GovernanceMetric metric = param.getMetric();
        Integer fp = JSON.parseObject(metric.getMetricParamsJson()).getInteger("file_permission");
        Integer dp = JSON.parseObject(metric.getMetricParamsJson()).getInteger("dir_permission");

        //获取当前表的路径
        String tableFsPath = param.getTableMetaInfo().getTableFsPath();

        //创建一个对象，保存检查的结果
        JSONObject jsonObject = new JSONObject();

        FileSystem hdfs = FileSystem.get(new URI(hdfsUri), new Configuration(), hdfsAdmin);

        //当前表目录
        FileStatus fileStatus = hdfs.getFileStatus(new Path(tableFsPath));
        checkPath(hdfs,fileStatus,jsonObject,fp,dp);

        //根据结果判断
        if (jsonObject.containsKey("wrong")){
            assessZeroScore(detail,jsonObject.getString("msg"),param,false);
        }
    }

    private void checkPath(FileSystem hdfs,FileStatus fileStatus,JSONObject jsonObject,Integer fp,Integer dp) throws Exception {
        if (fileStatus.isDirectory()){
            //根据目录的权限阈值进行检测
            checkPermission(fileStatus,dp,jsonObject);
            //当前目录是否已经超过权限，已经超过了，无需再继续了
            if (jsonObject.containsKey("wrong")){
                return;
            }
            //继续迭代目录中的文件，再判断
            FileStatus[] subFileStatus = hdfs.listStatus(fileStatus.getPath());
            for (FileStatus status : subFileStatus) {
                checkPath(hdfs,status,jsonObject,fp,dp);
                //如果有一个子文件超标了，没有必要再继续检测了
                if (jsonObject.containsKey("wrong")){
                    return;
                }
            }
        }else {
            //是文件的话，根据文件的阈值进行检测
            checkPermission(fileStatus,fp,jsonObject);
            if (jsonObject.containsKey("wrong")){
                return;
            }
        }
    }

    //用来根据类型，检测权限是否超过阈值
    private void checkPermission(FileStatus fileStatus,Integer allowPermission,JSONObject jsonObject){
        FsPermission permission = fileStatus.getPermission();
        //把 rw 转换为 数字
        int currentPermission = Integer.parseInt("" + permission.getUserAction().ordinal() + permission.getGroupAction().ordinal() + permission.getOtherAction().ordinal());

        if (currentPermission > allowPermission){
            jsonObject.put("wrong",true);
            jsonObject.put("msg","当前文件的权限是" + currentPermission + "，超过了要求的权限：" + allowPermission);
        }
    }
}
